package org.huangrui.spark.scala.core.req

import org.apache.spark.{SparkConf, SparkContext}

/**
 * @Author hr
 * @Create 2024-10-18 17:41 
 */
object HotCategoryTop10Analysis {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("spark")
    val sc = new SparkContext(conf)
    // TODO 将文件作为数据源，对接RDD进行操作
    val data = sc.textFile("data/user_visit_action.txt")


    // TODO 开发原则
    //      大数据开发：数据量大
    //      1. 多什么，删什么（减小数据规模）
    //          去除搜索数据
    //          对于的数据字段
    //      2. 缺什么，补什么
    //          品类。行为类型
    //      3. 功能实现中，要尽可能少地使用shuffle操作。shuffle越多，性能越低。
    // TODO 1. 将多余的数据进行删除（过滤）处理
    val filterRdd = data.filter((line: String) => {
      val strings = line.split("_")
      "null".matches(strings(5))
    })
    // TODO 2. 将过滤后的数据进行分组统计
    // -------------------------- 点击数量统计 -----------------------------------------
    // 保留点击数据
    val clickRdd = filterRdd.filter((line: String) => {
      val splits = line.split("_")
      splits(6) != "null"
    }).map((line: String) => {
      val splits = line.split("_")
      (splits(6), 1)
    }).reduceByKey(_ + _)
    // -------------------------- 下单数量统计 -----------------------------------------
    val orderRdd = filterRdd.filter((line: String) => {
      val splits = line.split("_")
      splits(8) != "null"
    }).flatMap((line: String) => {
      val splits = line.split("_")
      splits(8).split(",").map((item: String) => {
        (item, 1)
      })
    }).reduceByKey(_ + _)
    // -------------------------- 支付数量统计 -----------------------------------------
    val payRdd = filterRdd.filter((line: String) => {
      val splits = line.split("_")
      splits(10) != "null"
    }).flatMap((line: String) => {
      val splits = line.split("_")
      splits(10).split(",").map((item: String) => {
        (item, 1)
      })
    }).reduceByKey(_ + _)


    // 5. 将品类进行排序，并且取前10名
    //    点击数量排序，下单数量排序，支付数量排序
    //    元组排序：先比较第一个，再比较第二个，再比较第三个，依此类推
    //    ( 品类ID, ( 点击数量, 下单数量, 支付数量 ) )
    //
    //  cogroup = connect + group
    val cogRdd = clickRdd.cogroup(orderRdd, payRdd)
    val analysisRdd = cogRdd.mapValues {
      case (clicks, orders, pays) => {
        var totalClicks = 0
        val iter1 = clicks.iterator
        while (iter1.hasNext) {
          totalClicks += iter1.next()
        }
        var totalOrders = 0
        val iter2 = orders.iterator
        while (iter2.hasNext) {
          totalOrders += iter2.next()
        }
        var totalPays = 0
        val iter3 = pays.iterator
        while (iter3.hasNext) {
          totalPays += iter3.next()
        }
        (totalClicks, totalOrders, totalPays)
      }
    }
    analysisRdd.sortBy(_._2, false).take(10).foreach(println)
    sc.stop()
  }
}
